package com.patsnap.data.npd.dw.etl.utils;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @author lianbin
 * @date 2024/8/7 16:24
 * @desc 1.0
 */
public class KafkaUtils {

    public static KafkaProducer initKafkaProducer(String address) {
        Properties props = new Properties();
        props.put("bootstrap.servers", address); //服务器ip:端口号，集群用逗号分隔
        props.put("acks", "-1");
        props.put("retries", 3);
        props.put("max.request.size", 10485760);
        props.put("batch.size", 64 * 1024);
        props.put("buffer.memory", 64 * 1024 * 1024);
        props.put("linger.ms", 0);
        props.put("compression.type", "lz4");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        return new KafkaProducer(props);
    }

    public static void close(KafkaProducer producer) {
        if (null != producer) {
            producer.close();
        }
    }

    public static void sendMessage(KafkaProducer kafkaProducer, String topic, String message) throws ExecutionException, InterruptedException {
        ProducerRecord producerRecord = new ProducerRecord(topic, null, System.currentTimeMillis(), null, message);
        kafkaProducer.send(producerRecord).get();
    }
}
